package com.tcoiss.datafactory.service.mq;


import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;

import com.tcoiss.common.component.api.RemoteApiService;
import com.tcoiss.common.core.domain.R;
import com.tcoiss.common.mq.bean.IMdmReceiveHandle;
import com.tcoiss.common.mq.bean.MqMessage;
import com.tcoiss.common.mq.exception.MqException;
import com.tcoiss.common.component.api.model.SchemeVO;
import com.tcoiss.datafactory.domain.MqSyncRecord;
import com.tcoiss.datafactory.domain.MqTopic;
import com.tcoiss.datafactory.service.IExecuteSchemeService;
import com.tcoiss.datafactory.service.IMqSyncRecordService;
import com.tcoiss.datafactory.service.IMqTopicService;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.Date;
import java.util.List;
import java.util.stream.Collectors;

/**
 * @ClassName MdmReceiveHandleImpl
 * @Description 接收消息逻辑处理实现类
 * @author zhf
 * @Date 2017年9月20日 上午9:16:34
 * @version 1.0.0
 */
@Service
public class MdmReceiveHandleImpl implements IMdmReceiveHandle {

    private final Logger logger = LoggerFactory.getLogger(MdmReceiveHandleImpl.class);

    @Autowired
    private IExecuteSchemeService iExecuteSchemeService;

    @Resource
    private RemoteApiService remoteApiService;

    @Autowired
    private IMqTopicService iMqTopicService;
    @Autowired
    private IMqSyncRecordService syncRecordService;

    /**
     * @Description 判断是否执行逻辑代码
     * @param msgId
     * @param
     * @param tags
     * @param mdmMessage
     * @return
     */
    @Override
    public boolean doFilter(String msgId, String topicName, String tags, MqMessage mdmMessage) {
        //目前只是判断去重
        return !syncRecordService.selectIsExists(topicName, msgId);
    }



    /**
     * @Description 接收消息后执行代码
     * @param msgId
     * @param
     * @param tags
     * @param mqMessage
     * @throws MqException
     */
    @Override
    public void execute(String msgId, String topic, String tags, MqMessage mqMessage) throws MqException {
        // 根据通过主题查询数据工厂查询执行方案
        logger.info("执行处理逻辑，消息ID:" + msgId);
        try {
            R r = null;
            if(StringUtils.equals("saleorder",topic)){
                r = remoteApiService.sendWebSocket(mqMessage.getObjId());
                if(r==null){
                    throw new MqException("502",new Object[]{topic},"访问WebService服务异常", true);
                }
            }else {
                SchemeVO vo = new SchemeVO();
                vo.setExecuteNumber(topic);
                vo.setExecuteType(2);
                vo.setMsgObjectId(mqMessage.getObjId());
                r = iExecuteSchemeService.execute(vo);
                if(r==null){
                    throw new MqException("502",new Object[]{topic},"访问数据工厂异常", true);
                }
            }
            if(r.getCode()==200){
                if(topic!=null){
                    // 保存同步记录
                    MqSyncRecord record = new MqSyncRecord();
                    record.setMsgId(msgId);
                    record.setMsgBody(mqMessage.getJsonObject().toJSONString());
                    //record.setOperMethod(mdmMessage.getOperMethod().getValue());
                    record.setTopicCode(topic);
                    record.setTopicName(getTopicByCode(topic).getTopicName());
                    record.setSyncTime(new Date());
                    syncRecordService.save(record);
                }else{
                    throw new MqException("405",new Object[]{topic},"主题["+topic+"]执行方案失败: "+r.getMsg(), true);
                }
            }else{
                throw new MqException(r.getCode()+"",new Object[]{topic},"主题["+topic+"]执行方案失败: "+r.getMsg(), false);
            }
        } catch (Exception e) {
            throw new MqException("500",new Object[]{topic},"程序异常："+e.getMessage(), false);
        }
    }

    @Override
    public List<String> getTopics() {
        MqTopic topic = new MqTopic();
        topic.setDataLevel(1);//过滤已发布的数据
        List<MqTopic> mqTopics = iMqTopicService.queryList(topic);
        List<String> topics = mqTopics.stream().map(mqTopic ->mqTopic.getTopicCode()).collect(Collectors.toList());
        return topics;
    }

    private MqTopic getTopicByCode(String topicCode) {
        LambdaQueryWrapper<MqTopic> lqw = Wrappers.lambdaQuery();
        lqw.eq(MqTopic::getTopicCode,topicCode);
        return iMqTopicService.getOne(lqw);
    }
}
